package com.innovation.ic.sc.base.thread.listener.rabbitmq;

import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Strings;
import com.innovation.ic.sc.base.model.sc.ActionMessage;
import com.innovation.ic.sc.base.pojo.constant.handler.RabbitMqConstants;
import com.innovation.ic.sc.base.service.sc.ActionMessageService;
import com.innovation.ic.sc.base.vo.ActionMessageVo;
import com.rabbitmq.client.*;
import lombok.SneakyThrows;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.List;

/**
 * @desc   监听erp2sc的动作消息数据队列
 * @author linuo
 * @time   2022年10月18日15:37:22
 */
public class ListenActionMessageQueueThread extends AbstractRabbitmqThread implements Runnable {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    private Channel channel;
    private ActionMessageService actionMessageService;

    public ListenActionMessageQueueThread(Channel channel, ActionMessageService actionMessageService) {
        this.channel = channel;
        this.actionMessageService = actionMessageService;
    }

    @SneakyThrows
    public void run() {
        //String queue = RabbitMqConstants.ERP_2_SC_ACTION_MESSAGE_QUEUE;
        String exchange = RabbitMqConstants.ERP_2_SC_ACTION_MESSAGE_EXCHANGE;
        String routingKey = RabbitMqConstants.ERP_2_SC_ACTION_MESSAGE_ROUTING_KEY;
        final String queue = routingKey;
        channel.queueBind(queue, exchange, routingKey);
        channel.queueDeclare(queue, true, false, false, null);
        channel.exchangeDeclare(exchange, RabbitMqConstants.DIRECT_TYPE, true);
        Consumer callback = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String bodyString = new String(body);
                logger.info("接收到的队列:[{}]的消息为:[{}]", queue, bodyString);
                try {
                    if (!Strings.isNullOrEmpty(bodyString)) {
                        ActionMessage actionMessage = new ActionMessage();
                        ActionMessageVo actionMessageVo = JSONObject.parseObject(bodyString, ActionMessageVo.class);
                        if (actionMessage != null) {
                            List<String> createUserId = actionMessageVo.getCreateUserId();
                            actionMessage.setContent(actionMessageVo.getContent());
                            if(!CollectionUtils.isEmpty(createUserId)){
                                Assert.isTrue(actionMessageService.handleActionMessage(actionMessage, createUserId).getSuccess(),
                                        "handleErp2ScActionMessage异常：{createUserId}");
                            }else {
                                Assert.isTrue(actionMessageService.handleActionMessageByUserName(actionMessage,actionMessageVo.getCompanyName()).getSuccess(),
                                        "handleErp2ScActionMessage异常: {userName}");
                            }

                        }
                    }
                } catch (Exception e) {
                    logger.error("监听队列:[{}]时出现问题,原因:", queue, e);
                }
            }
        };
        channel.basicConsume(queue, true, callback);
    }
}